package com.alibaba.otter.manager.web.service;

import com.alibaba.otter.manager.biz.config.channel.ChannelService;
import com.alibaba.otter.manager.biz.statistics.delay.DelayStatService;
import com.alibaba.otter.manager.biz.statistics.delay.param.DelayStatInfo;
import com.alibaba.otter.manager.biz.statistics.delay.param.TopDelayStat;
import com.alibaba.otter.manager.biz.statistics.throughput.ThroughputStatService;
import com.alibaba.otter.manager.web.dto.DelayStatInfoDTO;
import com.alibaba.otter.manager.web.dto.TopDelayStatDTO;
import com.alibaba.otter.manager.web.mapstruct.DelayStatMapper;
import com.alibaba.otter.shared.arbitrate.ArbitrateManageService;
import com.alibaba.otter.shared.arbitrate.ArbitrateViewService;
import com.alibaba.otter.shared.arbitrate.model.MainStemEventData;
import com.alibaba.otter.shared.common.model.config.channel.Channel;
import com.alibaba.otter.shared.common.model.config.channel.ChannelStatus;
import com.alibaba.otter.shared.common.model.statistics.throughput.ThroughputStat;
import com.alibaba.otter.shared.common.model.statistics.throughput.ThroughputType;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

@Slf4j
@Service
public class DelayStatAggService {

    private final ThroughputStatService throughputStatService;

    private final DelayStatService delayStatService;

    private final ChannelService channelService;

    private final DelayStatMapper delayStatMapper;

    private final ArbitrateViewService arbitrateViewService;

    private final ArbitrateManageService arbitrateManageService;

    public DelayStatAggService(ThroughputStatService throughputStatService,
                               DelayStatService delayStatService,
                               ChannelService channelService,
                               DelayStatMapper delayStatMapper,
                               ArbitrateViewService arbitrateViewService, ArbitrateManageService arbitrateManageService) {
        this.throughputStatService = throughputStatService;
        this.delayStatService = delayStatService;
        this.channelService = channelService;
        this.delayStatMapper = delayStatMapper;
        this.arbitrateViewService = arbitrateViewService;
        this.arbitrateManageService = arbitrateManageService;
    }


    public DelayStatInfoDTO delayStat(Instant startTime, Instant endTime, Long pipelineId) {
        if (startTime == null || endTime == null) {
            startTime = Instant.ofEpochMilli(System.currentTimeMillis() / 60000 * 60000 - 24 * 60 * 60 * 1000);
            endTime = Instant.ofEpochMilli(System.currentTimeMillis() / 60000 * 60000);
        }

        Channel channel = channelService.findByPipelineId(pipelineId);
        Map<Long, DelayStatInfo> delayStatInfos = delayStatService.listTimelineDelayStat(pipelineId, startTime, endTime);

        Double delayAvg = 0.0;
        for (DelayStatInfo info : delayStatInfos.values()) {
            delayAvg += info.getAvgDelayTime();
        }

        if (!delayStatInfos.isEmpty()) {
            delayAvg = delayAvg / (1.0 * delayStatInfos.size());
        }

        Map<Long, DelayStatInfoDTO.RecentDelayStat> recentDelayStat = delayStatMapper.toRecentDelayStat(delayStatInfos);
        DelayStatInfoDTO delayStat = new DelayStatInfoDTO();
        delayStat.setDelayStatInfos(recentDelayStat);
        delayStat.setDelayAvg(delayAvg);
        delayStat.setChannel(channel);
        delayStat.setPipelineId(pipelineId);
        delayStat.setStartTime(startTime);
        delayStat.setEndTime(endTime);

        return delayStat;
    }


    public List<TopDelayStatDTO> topN(String searchKey, Integer topN, Long minute) {
        try {
            if (topN <= 0) {
                topN = 10;
            }

            if (minute <= 0) {
                minute = 1L;
            }

            List<TopDelayStat> tops = delayStatService.listTopDelayStat(searchKey, topN);

            List<Long> pipelineIds = new ArrayList<>();
            for (TopDelayStat top : tops) {
                top.setStatTime(minute);
                pipelineIds.add(top.getPipelineId());
            }

            List<ThroughputStat> stats = throughputStatService.listRealtimeThroughputByPipelineIds(pipelineIds, minute);
            for (ThroughputStat stat : stats) {
                for (TopDelayStat top : tops) {
                    if (stat.getPipelineId().equals(top.getPipelineId())) {
                        TopDelayStat.DataStat s = new TopDelayStat.DataStat(stat.getNumber(), stat.getSize());
                        if (ThroughputType.FILE == stat.getType()) {
                            top.setFileStat(s);
                        } else if (ThroughputType.ROW == stat.getType()) {
                            top.setDbStat(s);
                        }
                        break;
                    }
                }
            }

            List<TopDelayStatDTO> topDelayStat = delayStatMapper.toTopDelayStat(tops);
            for (TopDelayStatDTO top : topDelayStat) {

                ChannelStatus status = arbitrateManageService.channelEvent().status(top.getChannelId());
                MainStemEventData mainStemEventData = arbitrateViewService.mainstemData(top.getChannelId(), top.getPipelineId());
                top.setStatus(status);
                top.setEventData(mainStemEventData);
            }

            return topDelayStat;
        } catch (Exception e) {
            String errorMsg = String.format("error happens while searchKey[%s] topN [%d]", searchKey, topN);
            log.error(errorMsg, e);
        }
        return null;
    }
}
